package com.gpj.learn.rocketmq.simple;

import com.gpj.learn.rocketmq.CommonConstants;
import com.gpj.learn.rocketmq.base.BaseConsumer;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;

import java.util.*;

/**
 * @className SimpleConsumer
 * @description
 *   简单的消息消费者,包含[拉取模式的消费者]和[推送模式的消费者]
 * @author GPJ
 * @date 2021/4/22 21:58
 * @version 1.0
 **/
@Slf4j
public class SimpleConsumer {
	private static final Map<MessageQueue, Long> offset = new HashMap();

	public static void main(String[] args) throws Exception {
		SimpleConsumer consumer = new SimpleConsumer();
//		consumer.pull();
		consumer.push();
	}

	/**
	 * 拉取消息:定时从服务端拉取消息
	 */
	public void pull() throws Exception {
		DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("SimplePullConsumerGroupName");
		consumer.setNamesrvAddr(CommonConstants.ROCKETMQ_NAMESERVER_ADDR);
		consumer.start();
		Set<MessageQueue> queueSet = consumer.fetchSubscribeMessageQueues("TestTopic");
		while (true){
			for (MessageQueue queue : queueSet) {
				PullResult result = consumer.pullBlockIfNotFound(queue,
						null, offset(queue), 10);
				log.info("拉取到消息:{}", result);
				offset.put(queue, result.getNextBeginOffset());
			}
		}
	}

	private long offset(MessageQueue queue){
		Long offsetNum = offset.get(queue);
		return offsetNum == null?0:offsetNum;
	}

	/**
	 * 推送消息:服务端会主动推送消息到消息消费者
	 * @throws Exception
	 */
	public void push() throws Exception {
		DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("SimplePushConsumerGroupName");
		consumer.setNamesrvAddr(CommonConstants.ROCKETMQ_NAMESERVER_ADDR);
		consumer.subscribe("TestTopic", "*");
		consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
		consumer.setConsumeTimestamp("20210421221800");
		consumer.registerMessageListener((MessageListenerConcurrently) (list, context) -> {
			for (MessageExt message : list) {
				log.info("接收到消息:{}", message);

			}
			return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
		});
		consumer.start();
		log.info("consumer starting...");
	}
}
